package com.lianyi.rabbitdemo;

import android.annotation.SuppressLint;
import android.app.Service;
import android.content.Intent;
import android.os.Bundle;
import android.os.Handler;
import android.os.IBinder;
import android.os.Message;
import android.support.annotation.Nullable;
import android.util.Log;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.QueueingConsumer;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ReceiverService extends Service {
    private ConnectionFactory factory = new ConnectionFactory();
    private Thread subscribeThread;
    private Connection connection;

    @Nullable
    @Override
    public IBinder onBind(Intent intent) {
        return null;
    }

    @Override
    public void onCreate() {
        super.onCreate();
    }

    @Override
    public int onStartCommand(Intent intent, int flags, int startId) {
        Log.e("ASdcas", "onStartCommand: ");
        //连接设置
        setupConnectionFactory();
        //用于从线程中获取数据，更新ui
        @SuppressLint("HandlerLeak") final Handler incomingMessageHandler = new Handler() {
            @Override
            public void handleMessage(Message msg) {
                String message = msg.getData().getString("msg");
                Log.e("test", "msg:" + message);
            }
        };
        //开启消费者线程
        subscribe(incomingMessageHandler);
        return START_STICKY;
    }

    /**
     * 连接设置
     */
    private void setupConnectionFactory() {
        factory.setHost("192.168.6.102");//主机地址：
        factory.setPort(5672);// 端口号
        factory.setUsername("guest");// 用户名
        factory.setPassword("guest");// 密码
        factory.setVirtualHost("/");
        factory.setRequestedHeartbeat(1);
        factory.setAutomaticRecoveryEnabled(true);// 设置连接恢复
    }

    /**
     * 消费者线程
     */
    void subscribe(final Handler handler) {
        subscribeThread = new Thread(new Runnable() {
            @Override
            public void run() {
//                while (true) {
//                    try {
//                        // 需要再次初始化数据的时候就关闭上一个连接
//                        if (connection != null) {
//                            connection.close();
//                        }
//                        //使用之前的设置，建立连接
//                        connection = factory.newConnection();
//                        //创建一个通道
//                        Channel channel = connection.createChannel();
//                        //一次只发送一个，处理完成一个再获取下一个
//                        channel.basicQos(1);
//                        // 声明队列
//                        AMQP.Queue.DeclareOk q = channel.queueDeclare();
////                        channel.exchangeDeclare("fanout_test", "fanout", true);
//                        //2.声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
////                        channel.queueDeclare("queue", false, false, false, null);
//                        //将队列绑定到消息交换机exchange上
//                        //                  queue         exchange              routingKey路由关键字，exchange根据这个关键字进行消息投递。
////                        channel.queueBind(q.getQueue(), "fanout_test", "test_amq");
//                        QueueingConsumer consumer = new QueueingConsumer(channel);
//                        channel.basicConsume("queue", true, consumer);
//
//                        while (true) {
//                            try {
//                                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
//                                String routingKey = delivery.getEnvelope().getRoutingKey();
//                                String message = new String(delivery.getBody());
//
//                                Log.e("json_message", "---- " + message + " " + routingKey);
//
//                                Message msg = handler.obtainMessage();
//                                Bundle bundle = new Bundle();
//                                bundle.putString("json_message", message);
//                                msg.setData(bundle);
////                                handler.sendMessage(msg);
//                            } catch (InterruptedException ie) {
//                                ie.printStackTrace();
//                                return;
//                            }
//                        }
//
//                    } catch (Exception e1) {
//                        Log.d("", "Connection broken: " + e1.getClass().getName());
//                        try {
//                            Thread.sleep(5000); //sleep and then try again
//                        } catch (InterruptedException e) {
//                            break;
//                        }
//                    }
//                }
//                while (true) {
                    text();
//                }
            }
        });
        subscribeThread.start();
    }


    public void text() {
        try {
            // 需要再次初始化数据的时候就关闭上一个连接
            if (connection != null) {
                connection.close();
            }
            // 创建新的连接
            connection = factory.newConnection();
            // 创建通道
            final Channel channel = connection.createChannel();
            //在消息确认前，不接收其他消息, 处理完一个消息，再接收下一个消息
            channel.basicQos(1);
            channel.queueDeclare("send_test", true, false, false, null);
            channel.queueBind("", "e1", "send_test");
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String message = new String(body);
                    Log.e("message", "handleDelivery: " + message);
                    String routingKey = envelope.getRoutingKey();
                    String contentType = properties.getContentType();
                    long deliveryTag = envelope.getDeliveryTag();
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
            channel.basicConsume("send_test", false, consumer);

//            channel.basicConsume("queue", false, "", new DefaultConsumer(channel) {
//                @Override
//                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//                    super.handleDelivery(consumerTag, envelope, properties, body);
//                    String message = new String(body);
//                    Log.e("message", "handleDelivery: " + message);
//                    String routingKey = envelope.getRoutingKey();
//                    String contentType = properties.getContentType();
//                    long deliveryTag = envelope.getDeliveryTag();
//                    try {
//                        Thread.sleep(500);
//                    } catch (InterruptedException e) {
//                        e.printStackTrace();
//                    } finally {
//                        channel.basicAck(envelope.getDeliveryTag(), false);
//                    }
//                }
//            });
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void onDestroy() {
        super.onDestroy();
        subscribeThread.interrupt();
    }
}
